handler: (AsyncIterator[Datum]) -> Awaitable[list[BatchResponse]]

class SimpleBatchCat:
    async def handler(self, batch: AsyncIterator[Datum]) -> list[BatchResponse]:
        return [
            BatchResponse(
                d.id,
                [Message.to_drop()] if d.value == b"bad world" else [Message(d.value, keys=d.keys)],
            )
            async for d in batch
        ]


if __name__ == "__main__":
    batch_mapper_obj = SimpleBatchCat()
    batchmapper.BatchMapAsyncServer(
        batch_mapper_obj.handler,
        sock_file="/tmp/var/run/numaflow/batchmap.sock",
        server_info_file="/tmp/var/run/numaflow/mapper-server-info",
    ).run()

class SimpleBatchCat(batchmapper.BatchMapper):

class SimpleBatchCat:
    async def __call__(self, batch: AsyncIterator[Datum]) -> list[BatchResponse]:
        pass

batchmapper.BatchMapAsyncServer(
    SimpleBatchCat(),
).run()

def simple_batch_cat(drop_value: bytes):
    stats = {"processed": 0, "dropped": 0}

    async def handler(batch: AsyncIterator[Datum]) -> list[BatchResponse]:
        responses = []

        async for datum in batch:
            stats["processed"] += 1

            if datum.value == drop_value:
                stats["dropped"] += 1
                messages = [Message.to_drop()]
            else:
                messages = [Message(datum.value, keys=datum.keys)]

            responses.append(BatchResponse(datum.id, messages))

        print(f"Stats: {stats}")
        return responses

    return handler


batchmapper.BatchMapAsyncServer(
    simple_batch_cat(drop_value=b"bad world"),
).run()